package cn.doitedu.rtdw.data_dashboard;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/11
 * @Desc: 学大数据，到多易教育
 * 每小时，各品牌成交金额最高的前n个商品
 **/
public class Dash02_BrandTopnProducts {
    public static void main(String[] args) throws Exception {
        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setCheckpointTimeout(5000);
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.setParallelism(1);
        env.setStateBackend(new HashMapStateBackend());
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);



        // 创建cdc连接器表，映射业务库中的订单主表 oms_order

        tenv.executeSql(
                "CREATE TABLE order_mysql (    " +
                        "      id BIGINT," +
                        "      create_time TIMESTAMP(3),               " +
                        "      PRIMARY KEY (id) NOT ENFORCED           " +
                        "     ) WITH (                                 " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'realtimedw',          " +
                        "     'table-name' = 'oms_order'              " +
                        ")"
        );




        // 创建cdc连接器表，映射业务库中的订单详情表 oms_order_item
        tenv.executeSql(
                "CREATE TABLE item_mysql (    " +
                        "      id                       BIGINT,     " +
                        "      order_id                 BIGINT,     " +
                        "      product_id               BIGINT,     " +
                        "      product_brand            STRING,     " +
                        "      product_quantity         INT,        " +
                        "      product_price            DECIMAL,    " +
                        "      PRIMARY KEY(id) NOT ENFORCED         " +
                        "     ) WITH (                              " +
                        "     'connector' = 'mysql-cdc',            " +
                        "     'hostname' = 'doitedu'   ,            " +
                        "     'port' = '3306'          ,            " +
                        "     'username' = 'root'      ,            " +
                        "     'password' = 'root'      ,            " +
                        "     'database-name' = 'realtimedw',       " +
                        "     'table-name' = 'oms_order_item'       " +
                        ")"
        );


        // 对上述两表进行双流join
        tenv.executeSql(
                " CREATE TEMPORARY VIEW joined_view AS " +
                        "SELECT      "
                        +"   o.id as order_id      "
                        +"   ,o.create_time        " // join完后，此字段就失去了 *ROWTIME* 属性
                        +"   ,i.product_id         "
                        +"   ,i.product_brand      "
                        +"   ,i.product_quantity   "
                        +"   ,i.product_price      "
                        +" FROM item_mysql i       "
                        +" JOIN order_mysql o      "
                        +" ON o.id = i.order_id    "
        );

        //tenv.executeSql("select * from joined_view").print();

        // 将join后的结果表，先转成流
        DataStream<Row> joinedStream = tenv.toChangelogStream(tenv.from("joined_view"));
        joinedStream.print();

        /* 示意： 用api来处理流中的-u/+u/-d/+i等change语义
           // 根据你的意图，想怎么处理都能实现
        joinedStream.keyBy(new KeySelector<Row, Long>() {
            @Override
            public Long getKey(Row row) throws Exception {
                Long order_id = row.getFieldAs("order_id");
                return order_id;
            }
        }).window(TumblingEventTimeWindows.of(Time.hours(1)))
                .process(new ProcessWindowFunction<Row, Row, Long, TimeWindow>() {

                    HashMap<Long,Row> datas = new HashMap<Long,Row>();

                    @Override
                    public void process(Long aLong, ProcessWindowFunction<Row, Row, Long, TimeWindow>.Context context, Iterable<Row> rows, Collector<Row> out) throws Exception {

                        Iterator<Row> iterator = rows.iterator();

                        while (iterator.hasNext()){
                            Row row = iterator.next();
                            byte rowkind = row.getKind().toByteValue();
                            if(rowkind ==  0 || rowkind == 2){
                                Long order_id = row.getFieldAs("order_id");
                                row.setKind(RowKind.INSERT);
                                datas.put(order_id,row);
                            }
                        }

                        for (Row row : datas.values()) {

                            out.collect(row);
                        }

                    }
                })
        ;
        */

        // 然后再将流转成表，以便于在转表过程中，添加watermark及事件时间语义
        // tenv.createTemporaryView("tmp",joinedStream);  // 这样的转换无法消费-u/+u等change语义

        Table table = tenv.fromChangelogStream(joinedStream,
                Schema.newBuilder()
                        .column("order_id", DataTypes.BIGINT())
                        .column("create_time", DataTypes.TIMESTAMP(3))
                        .column("product_id", DataTypes.BIGINT())
                        .column("product_brand", DataTypes.STRING())
                        .column("product_quantity", DataTypes.INT())
                        .column("product_price", DataTypes.DECIMAL(10, 2))
                        .watermark("create_time", "create_time - interval '0' second")
                        .build()
        );
        tenv.createTemporaryView("tmp",table);


        // 按每小时滚动时间窗口进行聚合统计
        // TVF 聚合语法，底层所翻译出来的物理算子，不能接受CHANGELOG的-U/+U/-D语义
        // StreamPhysicalWindowAggregate doesn't support consuming update and delete changes
        /*tenv.executeSql(
                " SELECT                                                                    "
                        +"    window_start,                                                 "
                        +"    window_end,                                                   "
                        +"    product_brand,                                                "
                        +"    product_id,                                                   "
                        +"    sum(product_quantity * product_price) as amount               "
                        +" FROM TABLE (                                                     "
                        +" TUMBLE(TABLE tmp,DESCRIPTOR(create_time),INTERVAL '1' HOUR)      "
                        +" )                                                                "
                        +" GROUP BY                                                         "
                        +"    window_start,                                                 "
                        +"    window_end,                                                   "
                        +"    product_brand,                                                "
                        +"    product_id                                                    "
        ).print();*/

        // 用传统的 window组聚合语法，翻译出来的底层物理算子，可以接收各种changelog语义
        // 底层翻译的物理算子为： StreamPhysicalGroupWindowAggregate
        tenv.executeSql(
                " CREATE TEMPORARY VIEW summed  AS SELECT                                                                    "
                        +"    TUMBLE_start(create_time,INTERVAL '1' HOUR) as window_start,  "
                        +"    TUMBLE_end(create_time,INTERVAL '1' HOUR) as window_end,                                                   "
                        +"    product_brand,                                                "
                        +"    product_id,                                                   "
                        +"    sum(product_quantity * product_price) as amount               "
                        +" FROM tmp                                                         "
                        +" GROUP BY                                                         "
                        + "   TUMBLE(create_time,INTERVAL '1' HOUR),                        "
                        +"    product_brand,                                                "
                        +"    product_id                                                    "
        );

        // 求topn
        tenv.executeSql("SELECT\n" +
                "   window_start,\n" +
                "   window_end,\n" +
                "   product_brand,\n" +
                "   product_id,\n" +
                "   amount,\n" +
                "   rn\n" +
                "FROM (\n" +
                "SELECT\n" +
                "   window_start,\n" +
                "   window_end,\n" +
                "   product_brand,\n" +
                "   product_id,\n" +
                "   amount,\n" +
                "   row_number() over(partition by window_start,window_end,product_brand order by amount desc) as rn\n" +
                "FROM summed ) o\n" +
                "WHERE rn<=2").print();

        env.execute();

    }
}
